Vertex AI Pipelines

コードを実行したらコンパイルして Vertex AI Pipelines に投入する

手元から pipeline を投入するときに

code:run.py

if __name__ == "__main__":

# 邪魔じゃないなら単にこれでも

# dest = __file__.replace(".py", ".yaml")

import os

import tempfile

filename = os.path.basename(__file__).replace(".py", ".yaml")

dest = os.path.join(tempfile.mkdtemp(), filename)

compiler = kfp.compiler.Compiler()

compiler.compile(pipeline_func=pipeline, package_path=dest)

print(f"Generated pipeline: {dest}")

import google.cloud.aiplatform as aip

aip.init(project=project, location=location)

job = aip.PipelineJob(

display_name="sample",

template_path=dest,

)

job.submit(service_account="...")

マシンスペックを指定する

code:task_configurations.py

.add_accelerator_type

.set_accelerator_limit

.set_cpu_limit

.set_memory_request # Vertex AI Pipelines はサポートしてない

.set_memory_limit

.set_env_variable

.set_caching_options

.set_display_name

.set_retry

.ignore_upstream_failure

使いそうなやつ

.set_cpu_limit("4").set_memory_limit("16G")

.set_cpu_limit("2").set_memory_limit("16G")

.set_cpu_limit("4").set_memory_limit("15G")

.set_cpu_limit("4").set_memory_limit("26G")

GPU 指定

PipelineTask.add_node_selector_constraint() takes 2 positional arguments but 3 were given

あれ、最近の kfp (2.7)は key いらないらしい

print_torch().add_node_selector_constraint("NVIDIA_TESLA_T4").set_accelerator_limit(1)

GPU 付けたら起動までにめちゃめちゃ時間がかかる、10分~ 程度

リージョンやマシンタイプと組み合わせられる GPU に制限がある

低スペで GPU つけるとログ出ない

もっと目立つところに書いておけ

GPU で n1-highmem-2 などの小型のマシンタイプを使用すると、CPU の制約により、一部のワークロードでロギングが失敗する可能性があります。トレーニング ジョブがログを返さなくなった場合は、より大きなマシンタイプの選択を検討してください。

可能性がありますだけど、経験上 n1-highmem-2 で GPU 乗せるとコンポーネント内のログやエラーが残ったことない、デバッグにめちゃめちゃ苦労するので設定するのがよい

code:py

# n1-highmem-4 ($0.348910/h in Tokyo)

comp1()

.set_cpu_limit("4")

.set_memory_limit("26G")

.add_node_selector_constraint("NVIDIA_TESLA_T4")

.set_accelerator_limit(1)

# n1-standard-4 ($0.280600/h in Tokyo)

comp2()

.set_cpu_limit("4")

.set_memory_limit("15G")

.add_node_selector_constraint("NVIDIA_TESLA_T4")

.set_accelerator_limit(1)

Artifact の扱い

Artifact の名前を Output 以外にする

Pythonic artifact syntax で名前を指定する方法はない?

Cloud Storage に書き込む & Metadata 残す

独自型を定義してもよい、追加で出せる内容が増える

output = dsl.Dataset(uri=dsl.get_uri())

input.uri.replace("gs://", "/gcs/")

code:gcs_components.py

@dsl.component(base_image="python:3.11")

def save_artifact() -> dsl.Dataset:

out = dsl.Dataset(uri=dsl.get_uri(suffix='sample.txt')) # dsl.get_uri() で保存先を生成

with open(out.path, "w") as f: # Dataset に対応するローカルパスに書き込み

f.write("sample")

return out # Dataset 返す

@dsl.component(base_image="python:3.11")

def read_artifact(input: dsl.Dataset):

# gs://... が設定されているので GCS FUSE で読めるパスへ

# そのまま Cloud Storage Client などで読んでもよい

path = input.uri.replace("gs://", "/gcs/")

with open(path, "r") as f: # 開いて読む

print(f.readlines())

pass

@dsl.pipeline()

def pipeline():

save_artifact_task = save_artifact()

read_artifact(input=save_artifact_task.output)

Pipeline を Experiment と紐づける

code:experiment_linking.py

job = aip.PipelineJob(

template_path=dest,

parameter_values=parameter_values,

...

)

job.submit(

service_account=service_account,

experiment=experiment, # ここで名前つける

)

メトリックを記録する

pipeline の artifact として残る

Experiment に対しパイプラインが Experiment Run として紐づく

メトリックはパイプライン単位で残る、同じキーで書いてみても一応両方見れるが良くはなさそう

code:battle.py

@dsl.component(base_image="python:3.11")

metrics.log_metric("from", "a")

metrics.log_metric("a", 123)

@dsl.component(base_image="python:3.11")

metrics.log_metric("from", "b")

metrics.log_metric("b", 456)

パイプライン実行とテスト実行が同じテーブルなの微妙に嫌だけど、カラムあるってことは混ぜて使う想定かねえ

繰り返し実行で名前がかぶるので

被ってたら resume で上書きする

code:resume.py

name = "use-aiplatform-in-component"

resume = aiplatform.ExperimentRun.get(name) is not None

with aiplatform.start_run(name, resume=resume) as run:

run.log_metrics(...)

別々の ID を振る

task_id=dsl.PIPELINE_TASK_ID_PLACEHOLDER

code:task_id.py

name = f"use-aiplatform-in-component-{task_id}"

with aiplatform.start_run(name) as run:

run.log_metrics(...)

パイプライン実行リージョンと Exeperiment のリージョン別にする

kfp の metrics は使わず aiplatform で自分で送る、かなあ

GCS FUSE

仮想的に世界の全ての bucket がマウントされているような雰囲気、アクセスした時に権限がなければ弾かれる

impersonate で読み書きしたいが方法はなさそう

Secret Manager 使う

このサンプルの読み出すコンポーネントから受け渡したら Cloud Storage に残って良くないのでは?

pipeline 引数

キャッシュはデフォルト有効

らしいので 明に切る...と思ったけど実際あると便利

切る方法

パイプライン名、入出力、コンポーネントの外形的な仕様がキャッシュキーになる

パイプライン コンポーネントは決定論的に構築する必要があります。特定の入力セットでは、常に同じ出力が生成される必要があります

なのでコンポーネント内の実装いじって実行してもキャッシュのままなことがある

またコンポーネント実行で失敗していてもキャッシュ使い回される? ぽいので、修正しているつもりができてないことがある

コンポーネントとして呼ばれたパイプラインのキャッシュはどうなる?

共有の使い回しパイプライン

パイプラインA → 使い回しパイプライン

パイプラインB → 使い回しパイプライン

経験的にはされてなさそう(ちゃんとみてない)

親となる呼び出し元の名前が変わったらキャッシュは共有されない、まあそうか

キャッシュが効いていても出力するパイプライン実行には Artifact が追加される

助かる挙動

1. パイプライン実行1で Aritfact A を生成

2. パイプラインを修正

3. パイプライン実行2 で

Artifact A の生成はキャッシュが効いて処理をスキップ

Aritfact B を新たに生成

としたときに、A は 実行1, 実行2 両方のコンテキストから引ける、B は 実行2 からのみ

実行サービスアカウント

Failed to create pipeline job. Error: Vertex AI Service Agent ... does not have permission to access Artifact Registry repository

デフォルトでは Vertex AI Service Agent のサービスアカウントで動作する?

なのでカスタムイメージ pull するのに Artifact Registry の権限必要?

Artifact Registry からイメージを pull するようにカスタム サービス アカウントを構成することはできません。Vertex AI は、デフォルトのサービス アカウントを使用してイメージを pull します。

このデフォルトは Service Agent なのか、Compute Default なのかどっちだ

大抵実行用サービスアカウントを作って使うだろうし

失敗したコンポーネントが1つあれば全体を失敗させる

フェイルファストにする

Google Cloud 用コンポーネント

pipeline_root

省略するとデフォルトで以下が root になる

gs://{PROJECT_ID}-vertex-pipelines-{REGION}/output_artifacts/

各パイプラインが実行されるのは

{PIPELINE_ROOT}/{実行PROJECT_NUMBER?}/{PIPELINE_JOB_NAME}

既に存在する Artifact を返す

1つの func を別々の base_image を指定したい

コードコピペしたくねえが hermetic に(component の中に一通り)書くので使いまわしづらい...と思いきや

ここは普通の Python としてデコレータを単に呼べば良い

code:call_decorator.py

image_v1 = 'asia-northeast1-docker.pkg.dev/...'

image_v2 = 'asia-northeast1-docker.pkg.dev/...'

def my_component(...) -> dsl.Artifact:

...

foo_v1 = dsl.component(base_image=image_v1, func=my_component)

foo_v2 = dsl.component(base_image=image_v2, func=my_component)

@dsl.pipeline

def pipeline():

foo_v1_task = foo_v1(...)

foo_v2_task = foo_v2(...)

compare(v1=foo_v1_task.output, v2=foo_v2_task.output)